package com.niit.statis

import java.text.SimpleDateFormat
import java.util.Date

import org.apache.spark.SparkConf
import org.apache.spark.sql.{DataFrame, SparkSession}

// 定义评分表样本类
case class Rating(userId: Int, productId: Int, score: Double, timestamp: Int)
// 定义mongo配置类
case class MongoConfig(uri:String, db:String)

object StatisticsRecommender {

  val MONGODB_RATING_COLLECTION = "Rating" // 在initData模块中创建的评分表（userId,productId,score,timestamp）
  val RATE_MORE_PRODUCTS = "RateMoreProducts" // 定义历史热门商品表，通过计算商品的评分次数来推测
  val RATE_MORE_RECENTLY_PRODUCTS = "RateMoreRecentlyProducts" // 定义最近热门商品表（以月为单位进行统计）
  val AVERAGE_PRODUCTS = "AverageScoreProducts" // 定义商品的平均得分表（周期性计算）

  def main(args: Array[String]): Unit = {
    val config = Map( // 将mongodb配置信息用Map来定义，由于我的mongodb开启了权限认证，故需要指定用户名与密码
      "spark.cores" -> "local[*]",
      "mongo.uri" -> "mongodb://fooadmin:123456@hadoop001:27017/bigdata",
      "mongo.db" -> "bigdata"
    )
    // 创建SparkConf配置
    val sparkConf = new SparkConf().setAppName("StatisticsRecommender").setMaster(config("spark.cores"))
    // 创建SparkSession
    val spark = SparkSession.builder().config(sparkConf).getOrCreate()
    // 初始化MongoConfig,声明为implicit
    implicit  val mongoConfig = MongoConfig(config("mongo.uri"), config("mongo.db"))

    // 加入隐式转换
    import spark.implicits._

    // spark对象读取mongodb中的Rating表的数据，将其转换成DF
    val ratingDF = spark
      .read
      .option("uri", mongoConfig.uri)
      .option("collection", MONGODB_RATING_COLLECTION)
      .format("com.mongodb.spark.sql")
      .load()
      .as[Rating]
      .toDF()

    // 利用ratingDF创建一张名叫ratings的临时表（在SparkSession存活期间有效）
    ratingDF.createOrReplaceTempView("ratings")

    // TODO: 不同的统计推荐结果
    // 1. 历史热门商品统计
    statisticsHistory(spark)
    // 2. 近期热门商品
    recentlyHotProductsStatistics(spark)
    // 3. 商品平均得分
    averageProducts(spark)
    // 停止
    spark.stop()
  }

  // 1. 历史热门商品统计，按照商品的评分个数进行简单统计,并将统计结果写入到mongodb中
  def statisticsHistory(spark: SparkSession)(implicit mongoConfig: MongoConfig): Unit ={
    val historyDF = spark.sql("select productId, count(productId) as total from ratings group by productId order by total desc")
    insertIntoMongodb(historyDF,RATE_MORE_PRODUCTS)
  }

  // 2. 近期热门商品，把时间戳转换成yyyyMM格式进行评分个数统计，最终得到productId, count, yearmonth
  def recentlyHotProductsStatistics(spark: SparkSession)(implicit mongoConfig: MongoConfig): Unit ={
    // 2.1 创建一个时间格式化工具
    val simpleDateFormat = new SimpleDateFormat("yyyyMM")
    // 2.2 注册一个UDF函数，用于将timestamp装换成年月格式   1260759144000  => 201605
    spark.udf.register("formatDate",(x: Long) => simpleDateFormat.format(new Date( x * 1000L)).toLong)
    // 2.3 将原来的Rating数据集(即ratings)中的时间转换成年月的格式
    val dateFormatDF = spark.sql("select productId, score, formatDate(timestamp) as monthOfYear from ratings")
    // 2.4 将新的数据集dateFormatDF注册成一张临时表,如timeFormatRatings
    dateFormatDF.createOrReplaceTempView("timeFormatRatings")
    // 2.5 读取timeFormatRatings表中数据，按照productId和monthOfYear进行分组和排序，写sql交给spark.sql去实现即可
    val recentHotProductsDF = spark.sql("select productId, count(productId) as total, monthOfYear from timeFormatRatings group by monthOfYear,productId order by monthOfYear desc,total desc")
    // 2.6 调用方法写入mongo数据库表RateMoreRecentlyProducts中
    insertIntoMongodb(recentHotProductsDF,RATE_MORE_RECENTLY_PRODUCTS)
  }

  // 3. 商品平均得分：根据历史数据中所有用户对商品的评分，周期性的计算每个商品的平均得分,并将结果写入mongo中
  def averageProducts(spark: SparkSession)(implicit mongoConfig: MongoConfig): Unit ={
    val dataFrame = spark.sql("select productId, avg(score) as avgScore from ratings group by productId order by avgScore desc")
    insertIntoMongodb(dataFrame,AVERAGE_PRODUCTS)
  }

  // 自定义入库函数，用于将df存入mongodb中
  def insertIntoMongodb(df: DataFrame,collection_name: String)(implicit mongoConfig: MongoConfig): Unit ={
    df.write
      .option("uri",mongoConfig.uri)
      .option("collection",collection_name)
      .mode("overwrite")
      .format("com.mongodb.spark.sql")
      .save()
  }

}
